package org.etlt.job;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.etlt.EtltRuntimeException;
import org.etlt.check.CheckList;
import org.etlt.load.Loader;

import java.io.File;
import java.io.IOException;
import java.util.List;

/**
 * @see ParalleledJobExecutor
 */
public class JobExecutor {

    protected Log log = LogFactory.getLog(getClass());

    public void execute(JobContext context) {
        try {
            log.info("start job executing.");
            /**
             * execute all loaders
             */
            log.info("there are " + context.getAllLoaders().size() + " loaders to execute.");
            for (List<Loader> loaders = context.getUnfinishedLoaders();
                 !loaders.isEmpty();
                 loaders = context.getUnfinishedLoaders()) {
                log.info("this execution batch: " + loaders.size() + " loaders.");
                for (int i = 0; i < loaders.size(); i++) {
                    log.info("executing loader: " + (i + 1) + "/" + loaders.size());
                    Loader loader = loaders.get(i);
                    execute(context, loader);
                }
            }
            /**
             * execute all checklists
             */
            List<CheckList> checkLists = context.getAllChecklists();
            log.info("there are " + checkLists.size() + " checklists.");
            for (int i = 0; i < checkLists.size(); i++) {
                log.info("executing checklist: " + (i + 1) + "/" + checkLists.size());
                CheckList checkList = checkLists.get(i);
                execute(context, checkList);
            }
            ;
            log.info("job executing finished.");
        } finally {
            context.close();
        }
    }

    /**
     * execute designated loader
     *
     * @param loader
     */
    public void execute(JobContext jobContext, Loader loader) {
        try {
            log.info("loader executing: " + loader.getName());
            loader.preLoad(jobContext);
            loader.load(jobContext);
            loader.doFinish(jobContext);
            log.info("loader finished: " + loader.getName());
        }catch (Exception e){
            log.warn("load execution error.", e);
            jobContext.fail(loader);
            log.info("loader finished and failed："  + loader.getName() );
        }finally {
            try {
                loader.close();
            } catch (Exception e) {
                //do nothing
            }
        }
    }

    public void execute(JobContext context, CheckList checkList) {
        try {
            log.info("checklist starting:" + checkList.getName() );
            checkList.check(context);
            checkList.report(context);
            log.info("checklist finished:" + checkList.getName());
        }catch (Exception e){
            log.warn("checklist execution error.", e);
            log.info("checklist finished and failed："  + checkList.getName() );
        }
    }
}
